/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package mq

import (
	"fmt"
	"time"

	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
)

const (
	// ModeWorker worker(task)模式, 每个消息只被一个订阅的客户端消费
	ModeWorker Mode = "worker"

	// ModeTopic topic模式, 每个消息会被每个订阅的客户端消费(只支持异步发送，不支持消息回复等)
	ModeTopic Mode = "topic"
)

// Message TODO 结构体需要定义为protobuf类型的，方便后续在rabbitmq中处理
type Message struct {
	ID        string        `json:"id"` // 自动填充（无需手动指定）
	MainTopic string        `json:"mainTopic"`
	SubTopic  string        `json:"subTopic"` // 可以为空
	Ty        int64         `json:"ty"`
	Data      interface{}   `json:"data"`
	ChReply   chan *Message `json:"-"` // json 忽略该字段
	Err       error         `json:"-"`
}

// String print the message information
func (m *Message) String() string {
	return fmt.Sprintf("{id:%s, mTopic:%s, sTopic:%s, Ty:%s, Err:%v}", m.ID, m.MainTopic, m.SubTopic,
		types.GetEventName(int(m.Ty)), m.Err)
}

// Reply reply message to reply chan
func (m *Message) Reply(replyMsg *Message) {
	if m.ChReply == nil {
		// log.Debug("reply a empty ch reply", "msg", msg)
		return
	}

	replyMsg.ID = m.ID
	replyMsg.MainTopic = m.MainTopic
	replyMsg.SubTopic = m.SubTopic

	m.ChReply <- replyMsg
}

// Mode client mode
type Mode string

// String from mode to string
func (m Mode) String() string {
	return string(m)
}

// 每个模块都会有一个或者多个client对象
// 主要的操作大致如下：
// client := NewMqClient(mode)
// client.Sub("moduleName", "msgTopic")
// client.Sub("topic")
// async
// for msg := range client.Recv() {
//     process(msg)
// }
// sync
// client.send(msg, true)
// rsp = client.wait(msg, true)
// mode--client类型 (worker-task, topic)

// Client 消息队列接口
type Client interface {
	// 发送消息
	Send(msg *Message, sync bool) error
	SendTimeout(msg *Message, sync bool, timeout time.Duration) error

	// 等待应答消息(req--rsp)
	Wait(msg *Message) (*Message, error)
	WaitTimeout(msg *Message, timeout time.Duration) (*Message, error)

	// 接收订阅消息
	Recv() chan *Message

	// 订阅消息 subTopic可以为空 示例：Sub("moduleName", "msgTopic")
	Sub(mainTopic, subTopic string) error

	// 关闭client
	Close()

	// 新建消息
	NewMessage(mainTopic, subTopic string, ty int64, data interface{}) (msg *Message)
}

type funcProcess func(msg *Message)

// ProcessMsg process mq msg func
func ProcessMsg(msg *Message, cb funcProcess, client Client, log logging.Log) {
	beg := time.Now()
	ty := msg.Ty
	defer func() {
		log.Tracef("process msg ty %s cost time %v ", types.GetEventName(int(ty)), "cost", time.Since(beg))
		if r := recover(); r != nil {
			log.Errorf("processMsg recover panic error %v ", r)
			if client != nil {
				msg.Reply(client.NewMessage("", "", ty, fmt.Errorf("recover panic error:%v", r)))
			}
			return
		}
	}()
	cb(msg)
}
